package com.tjbklx1.redis.chap06;

import java.io.BufferedInputStream;
import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.FileWriter;
import java.io.FilenameFilter;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.io.Writer;
import java.util.ArrayDeque;
import java.util.Arrays;
import java.util.Deque;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.Set;
import java.util.zip.GZIPInputStream;
import java.util.zip.GZIPOutputStream;

import redis.clients.jedis.Jedis;

import com.tjbklx1.redis.JedisUtil;

public class App67_FileDistribution {

	public static void main(String[] args) throws Exception {
		new App67_FileDistribution().run();
	}

	private void run() throws Exception {
		Jedis conn = JedisUtil.getJedisWithDB();
		testFileDistribution(conn);
	}

	public void testFileDistribution(Jedis conn) throws InterruptedException, IOException {
		System.out.println("\n----- testFileDistribution -----");
		String[] keys = conn.keys("test:*").toArray(new String[0]);
		if (keys.length > 0) {
			conn.del(keys);
		}
		conn.del("msgs:test:", "seen:0", "seen:source", "ids:test:", "chat:test:");

		System.out.println("Creating some temporary 'log' files...");
		File f1 = File.createTempFile("temp_redis_1_", ".txt");
		f1.deleteOnExit();
		Writer writer = new FileWriter(f1);
		writer.write("one line\n");
		writer.close();

		File f2 = File.createTempFile("temp_redis_2_", ".txt");
		f2.deleteOnExit();
		writer = new FileWriter(f2);
		for (int i = 0; i < 100; i++) {
			writer.write("many lines " + i + '\n');
		}
		writer.close();

		File f3 = File.createTempFile("temp_redis_3_", ".txt.gz");
		f3.deleteOnExit();
		writer = new OutputStreamWriter(new GZIPOutputStream(new FileOutputStream(f3)));
		Random random = new Random();
		for (int i = 0; i < 1000; i++) {
			writer.write("random line " + Long.toHexString(random.nextLong()) + '\n');
		}
		writer.close();

		long size = f3.length();
		System.out.println("Done.");
		System.out.println();
		System.out.println("Starting up a thread to copy logs to redis...");
		File path = f1.getParentFile();
		CopyLogsThread thread = new CopyLogsThread(path, "test:", 1, size);
		thread.start();

		System.out.println("Let's pause to let some logs get copied to Redis...");
		Thread.sleep(250);
		System.out.println();
		System.out.println("Okay, the logs should be ready. Let's process them!");

		System.out.println("Files should have 1, 100, and 1000 lines");
		TestCallback callback = new TestCallback();
		processLogsFromRedis(conn, "0", callback);
		System.out.println(Arrays.toString(callback.counts.toArray(new Integer[0])));
		assert callback.counts.get(0) == 1;
		assert callback.counts.get(1) == 100;
		assert callback.counts.get(2) == 1000;

		System.out.println();
		System.out.println("Let's wait for the copy thread to finish cleaning up...");
		thread.join();
		System.out.println("Done cleaning out Redis!");

		keys = conn.keys("test:*").toArray(new String[0]);
		if (keys.length > 0) {
			conn.del(keys);
		}
		conn.del("msgs:test:", "seen:0", "seen:source", "ids:test:", "chat:test:");
	}

	public void processLogsFromRedis(Jedis conn, String id, Callback callback) throws InterruptedException, IOException {
		while (true) {
			List<ChatMessages> fdata = App66_MultiRecipientMessaging.fetchPendingMessages(conn, id);

			for (ChatMessages messages : fdata) {
				for (Map<String, Object> message : messages.messages) {
					String logFile = (String) message.get("message");

					if (":done".equals(logFile)) {
						return;
					}
					if (logFile == null || logFile.length() == 0) {
						continue;
					}

					InputStream in = new RedisInputStream(conn, messages.chatId + logFile);
					if (logFile.endsWith(".gz")) {
						in = new GZIPInputStream(in);
					}

					BufferedReader reader = new BufferedReader(new InputStreamReader(in));
					try {
						String line = null;
						while ((line = reader.readLine()) != null) {
							callback.callback(line);
						}
						callback.callback(null);
					} finally {
						reader.close();
					}

					conn.incr(messages.chatId + logFile + ":done");
				}
			}

			if (fdata.size() == 0) {
				Thread.sleep(100);
			}
		}
	}

	public class CopyLogsThread extends Thread {
		private Jedis conn;
		private File path;
		private String channel;
		private int count;
		private long limit;

		public CopyLogsThread(File path, String channel, int count, long limit) {
			this.conn = new Jedis("localhost");
			this.conn.select(15);
			this.path = path;
			this.channel = channel;
			this.count = count;
			this.limit = limit;
		}

		public void run() {
			Deque<File> waiting = new ArrayDeque<File>();
			long bytesInRedis = 0;

			Set<String> recipients = new HashSet<String>();
			for (int i = 0; i < count; i++) {
				recipients.add(String.valueOf(i));
			}
			App66_MultiRecipientMessaging.createChat(conn, "source", recipients, "", channel);
			File[] logFiles = path.listFiles(new FilenameFilter() {
				public boolean accept(File dir, String name) {
					return name.startsWith("temp_redis");
				}
			});
			Arrays.sort(logFiles);
			for (File logFile : logFiles) {
				long fsize = logFile.length();
				while ((bytesInRedis + fsize) > limit) {
					long cleaned = clean(waiting, count);
					if (cleaned != 0) {
						bytesInRedis -= cleaned;
					} else {
						try {
							sleep(250);
						} catch (InterruptedException ie) {
							Thread.interrupted();
						}
					}
				}

				BufferedInputStream in = null;
				try {
					in = new BufferedInputStream(new FileInputStream(logFile));
					int read = 0;
					byte[] buffer = new byte[8192];
					while ((read = in.read(buffer, 0, buffer.length)) != -1) {
						if (buffer.length != read) {
							byte[] bytes = new byte[read];
							System.arraycopy(buffer, 0, bytes, 0, read);
							conn.append((channel + logFile).getBytes(), bytes);
						} else {
							conn.append((channel + logFile).getBytes(), buffer);
						}
					}
				} catch (IOException ioe) {
					ioe.printStackTrace();
					throw new RuntimeException(ioe);
				} finally {
					try {
						in.close();
					} catch (Exception ignore) {
					}
				}

				App66_MultiRecipientMessaging.sendMessage(conn, channel, "source", logFile.toString());

				bytesInRedis += fsize;
				waiting.addLast(logFile);
			}

			App66_MultiRecipientMessaging.sendMessage(conn, channel, "source", ":done");

			while (waiting.size() > 0) {
				long cleaned = clean(waiting, count);
				if (cleaned != 0) {
					bytesInRedis -= cleaned;
				} else {
					try {
						sleep(250);
					} catch (InterruptedException ie) {
						Thread.interrupted();
					}
				}
			}
		}

		private long clean(Deque<File> waiting, int count) {
			if (waiting.size() == 0) {
				return 0;
			}
			File w0 = waiting.getFirst();
			if (String.valueOf(count).equals(conn.get(channel + w0 + ":done"))) {
				conn.del(channel + w0, channel + w0 + ":done");
				return waiting.removeFirst().length();
			}
			return 0;
		}
	}
}
